package broker

import (
	"context"
	"fmt"
	"github.com/go-kratos/kratos/v2/log"
	"github.com/tx7do/kratos-transport/broker"
	"github.com/tx7do/kratos-transport/broker/kafka"
	"github.com/tx7do/kratos-transport/broker/nsq"
	"github.com/tx7do/kratos-transport/broker/rocketmq"
	"kratos_kafka_new/internal/conf"
)

type Broker struct {
	conf   *conf.Bootstrap
	logger *log.Helper
	Kafka  broker.Broker
	//Nsq      broker.Broker
	//RocketMq broker.Broker
}

func NewBroker(conf *conf.Bootstrap, logger log.Logger) *Broker {
	return &Broker{
		conf:   conf,
		logger: log.NewHelper(logger),
		Kafka:  NewBootKafka(conf, logger).Run(),
		//Nsq: NewBootNsq(conf, logger).Run(),
		//RocketMq: NewBootRocket(conf, logger).Run(),
	}
}

func NewBootKafka(conf *conf.Bootstrap, logger log.Logger) *Kafka {
	return &Kafka{
		conf,
		log.NewHelper(logger),
	}
}

func NewBootNsq(conf *conf.Bootstrap, logger log.Logger) *Nsq {
	return &Nsq{
		conf,
		log.NewHelper(logger),
	}
}

func NewBootRocket(conf *conf.Bootstrap, logger log.Logger) *Rocket {
	return &Rocket{
		conf,
		log.NewHelper(logger),
	}
}

// kafka
type Kafka struct {
	conf   *conf.Bootstrap
	logger *log.Helper
}

func (b *Kafka) Run() broker.Broker {
	// TLS～～
	//certBytes, err := ioutil.ReadFile(b.conf.MessageQueue.Kafka.Pem)
	//if err != nil {
	//	panic(fmt.Sprintf("kafka client read cert file failed %v", err))
	//}
	//clientCertPool := x509.NewCertPool()
	//ok := clientCertPool.AppendCertsFromPEM(certBytes)
	//if !ok {
	//	panic("kafka client failed to parse root certificate")
	//}

	ctx := context.Background()
	kafkaBroker := kafka.NewBroker(
		broker.WithOptionContext(ctx),
		broker.WithAddress(b.conf.MessageQueue.Kafka.Broker.Addr...),
		broker.WithCodec("json"),
		// 发送批次大小
		kafka.WithBatchSize(1),
		// 异步发送消息
		kafka.WithAsync(true),

		// TLS～～
		//broker.WithTLSConfig(&tls.Config{
		//	RootCAs:            clientCertPool,
		//	InsecureSkipVerify: true,
		//}),
		//kafka.WithPlainMechanism(b.conf.MessageQueue.Kafka.Username, b.conf.MessageQueue.Kafka.Password),
	)
	_ = kafkaBroker.Init()
	if err := kafkaBroker.Connect(); err != nil {
		panic(fmt.Sprintf("Icant connect to broker, skip: %v", err))
	}
	fmt.Println("初始化kafka的broker成功!!!!!")
	return kafkaBroker
}

// nsq
type Nsq struct {
	conf   *conf.Bootstrap
	logger *log.Helper
}

func (b *Nsq) Run() broker.Broker {
	ctx := context.Background()
	nsqBroker := nsq.NewBroker(
		broker.WithOptionContext(ctx),
		broker.WithAddress(b.conf.MessageQueue.Nsq.Broker.Addr...),
	)
	_ = nsqBroker.Init()
	if err := nsqBroker.Connect(); err != nil {
		b.logger.WithContext(context.Background()).Errorw("reason", "broker nsq connect error", "error", err)
	}
	return nsqBroker
}

// rocketMQ
type Rocket struct {
	conf   *conf.Bootstrap
	logger *log.Helper
}

func (b *Rocket) Run() broker.Broker {
	ctx := context.Background()
	rocketBroker := rocketmq.NewBroker(
		broker.WithOptionContext(ctx),
		// rocketmq的配置
		rocketmq.WithAliyunHttpSupport(),
		rocketmq.WithEnableTrace(),
		// Notice 这些是阿里云 rocketMQ 的参数，在配置文件中
		rocketmq.WithNameServerDomain(b.conf.MessageQueue.Rocket.Broker.EndPoint),
		rocketmq.WithAccessKey(b.conf.MessageQueue.Rocket.Broker.AccessId),
		rocketmq.WithSecretKey(b.conf.MessageQueue.Rocket.Broker.AccessKey),
		rocketmq.WithInstanceName(b.conf.MessageQueue.Rocket.Broker.InstanceId),
	)
	_ = rocketBroker.Init()
	if err := rocketBroker.Connect(); err != nil {
		b.logger.WithContext(context.Background()).Errorw("reason", "broker rocket connect error", "error", err)
	}
	return rocketBroker
}
